package cn.fangline.wo.service;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.Terminated;
import akka.pattern.Patterns;
import akka.routing.RoundRobinPool;
import akka.util.Timeout;
import cn.fangline.wo.model.User;
import cn.fangline.wo.task.akka.actors.*;
import cn.fangline.wo.task.akka.messages.CheckAlive;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.google.common.util.concurrent.RateLimiter;
import com.typesafe.config.ConfigFactory;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Service;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;

import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

/**
 * Created by 抽离 on 2017/6/23 0023.
 */
@Service
public class AnjukeCrawler {

    private Logger logger = Logger.getLogger(AnjukeCrawler.class);

    public void start() throws Exception {
        long begin = System.currentTimeMillis();
        System.out.println("Anjuke Crawler Begin" + LocalDateTime.now());
        RateLimiter rateLimiter = RateLimiter.create(35.0);

        ActorSystem masterSystem = ActorSystem.create("masterSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem agentSystem = ActorSystem.create("agentSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem agentMergeSystem = ActorSystem.create("agentMergeSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem houseMasterSystem = ActorSystem.create("houseMasterSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem houseSystem = ActorSystem.create("houseSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem housePackSystem = ActorSystem.create("housePackSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem houseExistSystem = ActorSystem.create("houseExistSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem houseInsertSystem = ActorSystem.create("houseInsertSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem houseUpdateSystem = ActorSystem.create("houseUpdateSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem houseImageSystem = ActorSystem.create("houseImageSystem", ConfigFactory.load().getConfig("ActorSystem"));
        ActorSystem houseImageMasterSystem = ActorSystem.create("houseImageMasterSystem", ConfigFactory.load().getConfig("ActorSystem"));


        ActorRef houseImageRef = houseImageSystem.actorOf(Props.create(HouseImageActor.class, rateLimiter)
                        .withDispatcher("houseImageDispatcher")
                        .withRouter(new RoundRobinPool(400))
                         , "houseImageRef"
        );
        ActorRef houseImageMasterRef = houseImageMasterSystem.actorOf(Props.create(HouseImageMasterActor.class, houseImageRef)
                .withDispatcher("houseImageMasterDispatcher"), "houseImageMasterRef"
        );
        ActorRef houseInsertRef = houseInsertSystem.actorOf(Props.create(HouseInsertActor.class, houseImageMasterRef)
                .withDispatcher("defaultDispatcher"), "houseInsertRef"
        );
        ActorRef houseUpdateRef = houseUpdateSystem.actorOf(Props.create(HouseUpdateActor.class)
                .withDispatcher("defaultDispatcher"), "houseUpdateRef"
        );
        ActorRef houseExistRef = houseExistSystem.actorOf(Props.create(HouseExistActor.class, houseInsertRef, houseUpdateRef)
                .withDispatcher("defaultDispatcher"), "houseExistRef"
        );
        ActorRef housePackRef = housePackSystem.actorOf(Props.create(HousePackActor.class, houseExistRef)
                .withDispatcher("defaultDispatcher"), "housePackRef"
        );
        ActorRef houseRef = houseSystem.actorOf(Props.create(HouseActor.class, rateLimiter, housePackRef)
                        .withDispatcher("houseDispatcher")
                        .withRouter(new RoundRobinPool(400))
                        , "houseRef");
        ActorRef houseMasterRef = houseMasterSystem.actorOf(Props.create(HouseMasterActor.class, houseRef)
                .withDispatcher("houseMasterDispatcher"), "houseMasterRef");
        ActorRef agentMergerRef = agentMergeSystem.actorOf(Props.create(AgentMerger.class, houseMasterRef)
                .withDispatcher("agentMergeDispatcher"), "agentMergerRef");
        ActorRef agentRef = agentSystem.actorOf(Props.create(AgentActor.class, agentMergerRef)
                .withDispatcher("agentDispatcher"), "agentRef");
        ActorRef masterRef = masterSystem.actorOf(Props.create(MasterActor.class, agentRef)
                .withDispatcher("defaultDispatcher"), "masterRef");



        boolean isAlive = true;

        while (isAlive){
//            Future<Terminated> future = agentSystem.whenTerminated();
//            if(future.isCompleted()){
//                isAlive = false;
//            }
            Timeout timeout = new Timeout(Duration.create(20 , "seconds"));
            Future<Object> future = Patterns.ask(houseImageMasterRef, new CheckAlive(), timeout);
            isAlive = (Boolean) Await.result(future, timeout.duration());
            Thread.sleep(1000 * 20);
//            System.out.println("----------=" + (String)Await.result(future, timeout.duration()));
        }
        System.out.println("Anjuke Crawler End" + LocalDateTime.now());
        System.out.println("总共耗时：" + (System.currentTimeMillis() - begin)/(1000*60) + "分钟");
    }

    public void test(){
        System.out.println("----------=" + new User().selectList(new EntityWrapper<User>().in("corp_id", "11596")));
    }
}
